package com.gzhryc.common.kafka.consumer;

import com.gzhryc.common.kafka.JFKafkaConsumer;
import com.gzhryc.common.kafka.consumer.interfaces.KafkaConsumerEventListener;
import com.gzhryc.common.kafka.models.ConsumerEvent;
import com.gzhryc.common.logging.Logger;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 单个线程对一个主题进行消费的消费者
 *
 * @author zhanghao
 */
public class SingleKafkaConsumer implements JFKafkaConsumer {

    static Logger log = Logger.getLogger(SingleKafkaConsumer.class);

    private String topic;
    private String groupId;
    private KafkaConsumer<String, String> kafkaConsumer;
    private KafkaConsumerEventListener listener;

    private MainThread mainThread;
    private AtomicBoolean isRunning;
    private Long times = 10L;

    public SingleKafkaConsumer(String topic, KafkaConsumerEventListener listener) {
        this.topic = topic;
        this.groupId = topic;
        this.listener = listener;
        this.isRunning = new AtomicBoolean(false);
    }

    public SingleKafkaConsumer(String topic, String groupId, KafkaConsumerEventListener listener) {
        this.topic = topic;
        this.groupId = groupId;
        this.listener = listener;
        this.isRunning = new AtomicBoolean(false);
    }

    public void start(Properties properties) {
        if (properties != null && this.isRunning.compareAndSet(false, true)) {
            Properties temp = (Properties) properties.clone();
            temp.setProperty("group.id", this.groupId);
            this.mainThread = new MainThread(temp);
            this.mainThread.start();
            if (listener != null) {
                listener.start(topic);
            }
        }
    }

    public void close() {
        this.isRunning.set(false);
    }

    public boolean isRunning() {
        return this.isRunning.get();
    }

    public String getTopic() {
        return this.topic;
    }

    public String getGroupId() {
        return this.groupId;
    }

    public void setTimes(Long times) {
        this.times = times;
    }

    private class MainThread extends Thread {

        final Properties properties;

        public MainThread(Properties properties) {
            this.properties = properties;
        }

        public void run() {
            kafkaConsumer = new KafkaConsumer<String, String>(properties);
            kafkaConsumer.subscribe(Collections.singletonList(topic));
            while (isRunning.get()) {
                try {
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(times));
                    List<ConsumerEvent> events = new ArrayList<ConsumerEvent>();
                    for (ConsumerRecord<String, String> record : records) {
                        ConsumerEvent event = new ConsumerEvent();
                        event.setKey(record.key());
                        event.setTopic(record.topic());
                        event.setValue(record.value());
                        event.setPartition(record.partition());
                        event.setGroupId(properties.getProperty("group.id"));
                        events.add(event);
                    }

                    if (listener != null) {
                        try {
                            listener.receive(events);
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }

                    kafkaConsumer.commitAsync((offsets, e) -> {
                        if (e != null) {
                            log.error("kafka commit failed for" + offsets + ",e:" + e.getMessage());
                        }
                    }); // 手动异步提交( 重点 )
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
            // 关闭KafkaConsumer
            if (kafkaConsumer != null) {
                kafkaConsumer.close();
            }
            isRunning.set(false);
        }
    }
}
